#ifndef __M_BROKER_H__
#define __M_BROKER_H__

#include "../third/include/muduo/net/TcpServer.h"
#include "../third/include/muduo/net/EventLoop.h"
#include "../third/include/muduo/net/TcpConnection.h"
#include "../third/include/muduo/base/Logging.h"
#include "../third/include/muduo/base/Mutex.h"
#include "../third/include/muduo/proto/codec.h"
#include "../third/include/muduo/proto/dispatcher.h"

#include "connection.hpp"
#include "virtualHost.hpp"
#include "consumer.hpp"
#include "user.hpp"
#include <iostream>
#include <string>
#include <unordered_map>

namespace zhong
{

    class BrokerServer
    {
        BrokerServer(int port, const std::string &data_dir)
            : _server(&_base_loop, muduo::net::InetAddress("127.0.0.1", port), "Server", muduo::net::TcpServer::kReusePort),
              _dispatcher(std::bind(&BrokerServer::onUnknownMessage, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3)),
              _codec(std::make_shared<ProtobufCodec>(std::bind(&ProtobufDispatcher::onProtobufMessage, &_dispatcher, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3))),
              _host_manager(std::make_shared<VirtualHostManager>(data_dir + "virtual_host/")),
              _user_manager(std::make_shared<UserManager>(data_dir + "user/")),
              _connection_manager(std::make_shared<ConnectionManager>()),
              _consumer_manager(std::make_shared<ConsumerManager>()),
              _threadpool(std::make_shared<threadpool>())
        {

            // 2.设置连接、协议处理器的callback函数
            _server.setConnectionCallback(std::bind(&BrokerServer::OnConnection, this, std::placeholders::_1));

            _server.setMessageCallback(std::bind(&ProtobufCodec::onMessage, _codec.get(), std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));

            // 3.设置函数分发器注册callback函数
            _dispatcher.registerMessageCallback<zhong::openChannelRequest>(std::bind(&BrokerServer::onOpenChannel, this,
                                                                                     std::placeholders::_1,
                                                                                     std::placeholders::_2,
                                                                                     std::placeholders::_3));
            _dispatcher.registerMessageCallback<zhong::closeChannelRequest>(std::bind(&BrokerServer::onCloseChannel, this,
                                                                                      std::placeholders::_1,
                                                                                      std::placeholders::_2,
                                                                                      std::placeholders::_3));
            _dispatcher.registerMessageCallback<zhong::declareExchangeRequest>(std::bind(&BrokerServer::onDeclareExchange, this,
                                                                                         std::placeholders::_1,
                                                                                         std::placeholders::_2,
                                                                                         std::placeholders::_3));
            _dispatcher.registerMessageCallback<zhong::deleteExchangeRequest>(std::bind(&BrokerServer::onDeleteExchange, this,
                                                                                        std::placeholders::_1,
                                                                                        std::placeholders::_2,
                                                                                        std::placeholders::_3));
            _dispatcher.registerMessageCallback<zhong::declareQueueRequest>(std::bind(&BrokerServer::onDeclareQueue, this,
                                                                                      std::placeholders::_1,
                                                                                      std::placeholders::_2,
                                                                                      std::placeholders::_3));
            _dispatcher.registerMessageCallback<zhong::deleteQueueRequest>(std::bind(&BrokerServer::onDeleteQueue, this,
                                                                                     std::placeholders::_1,
                                                                                     std::placeholders::_2,
                                                                                     std::placeholders::_3));
            _dispatcher.registerMessageCallback<zhong::queueBindRequest>(std::bind(&BrokerServer::onQueueBind, this,
                                                                                   std::placeholders::_1,
                                                                                   std::placeholders::_2,
                                                                                   std::placeholders::_3));
            _dispatcher.registerMessageCallback<zhong::queueUnBindRequest>(std::bind(&BrokerServer::onQueueUnBind, this,
                                                                                     std::placeholders::_1,
                                                                                     std::placeholders::_2,
                                                                                     std::placeholders::_3));
            _dispatcher.registerMessageCallback<zhong::basicPublishRequest>(std::bind(&BrokerServer::onBasicPublish, this,
                                                                                      std::placeholders::_1,
                                                                                      std::placeholders::_2,
                                                                                      std::placeholders::_3));
            _dispatcher.registerMessageCallback<zhong::basicAckRequest>(std::bind(&BrokerServer::onBasicAck, this,
                                                                                  std::placeholders::_1,
                                                                                  std::placeholders::_2,
                                                                                  std::placeholders::_3));
            _dispatcher.registerMessageCallback<zhong::basicConsumeRequest>(std::bind(&BrokerServer::onBasicConsume, this,
                                                                                      std::placeholders::_1,
                                                                                      std::placeholders::_2,
                                                                                      std::placeholders::_3));
            _dispatcher.registerMessageCallback<zhong::basicCancelRequest>(std::bind(&BrokerServer::onBasicCancel, this,
                                                                                     std::placeholders::_1,
                                                                                     std::placeholders::_2,
                                                                                     std::placeholders::_3));
            _dispatcher.registerMessageCallback<zhong::declareHostRequest>(std::bind(&BrokerServer::onDeclareHost, this,
                                                                                     std::placeholders::_1,
                                                                                     std::placeholders::_2,
                                                                                     std::placeholders::_3));
            _dispatcher.registerMessageCallback<zhong::deleteHostRequest>(std::bind(&BrokerServer::onDeleteHost, this,
                                                                                    std::placeholders::_1,
                                                                                    std::placeholders::_2,
                                                                                    std::placeholders::_3));
            _dispatcher.registerMessageCallback<zhong::selectHostRequest>(std::bind(&BrokerServer::onSelectHost, this,
                                                                                    std::placeholders::_1,
                                                                                    std::placeholders::_2,
                                                                                    std::placeholders::_3));
            _dispatcher.registerMessageCallback<zhong::declareUserRequest>(std::bind(&BrokerServer::onDeclareUser, this,
                                                                                     std::placeholders::_1,
                                                                                     std::placeholders::_2,
                                                                                     std::placeholders::_3));
            _dispatcher.registerMessageCallback<zhong::deleteUserRequest>(std::bind(&BrokerServer::onDeleteUser, this,
                                                                                    std::placeholders::_1,
                                                                                    std::placeholders::_2,
                                                                                    std::placeholders::_3));
            _dispatcher.registerMessageCallback<zhong::selectUserRequest>(std::bind(&BrokerServer::onSelectUser, this,
                                                                                    std::placeholders::_1,
                                                                                    std::placeholders::_2,
                                                                                    std::placeholders::_3));
        }
        // 单例模式不允许赋值重载
        BrokerServer(const BrokerServer &LM) = delete;
        BrokerServer &operator=(const BrokerServer &LM) = delete;

    public:
        // 实现单例模式
        static BrokerServer *GetInstance(int port, const std::string &data_dir)
        {
            if (_instance == nullptr)
            {
                // LockGuard guard(&_mutex);
                if (_instance == nullptr)
                    _instance = new BrokerServer(port, data_dir);
            }

            return _instance;
        }
        // 服务器启动
        void start()
        {
            // 进行listen客户端链接
            _server.start();
            // Reactor设计模式下进行epoll的监控
            _base_loop.loop();
        }

    private:

        /**************************************   连接处理回调函数    *******************************************************/
        
        void onUnknownMessage(const muduo::net::TcpConnectionPtr &conn, const std::shared_ptr<google::protobuf::Message> &message, muduo::Timestamp)
        {
            ERR_LOG("收到未知请求[%s]", message->GetTypeName().c_str());
            conn->shutdown();
        }
        // 获取链接的回调函数
        void OnConnection(const muduo::net::TcpConnectionPtr &conn) // 链接建立时发生的回调函数
        {
            if (conn->connected() == true)
            {
                // 连接构建成功后实例化一个链接对象
                _connection_manager->createConnection(_consumer_manager, _host_manager, _user_manager, _codec, conn, _threadpool);
                INFO_LOG("获得一个新链接[%s]", conn->name().c_str());
            }
            else
            {
                _connection_manager->deleteConnection(conn);
                INFO_LOG("链接[%s]关闭", conn->name().c_str());
            }
        }

        /**************************************   业务处理回调函数    *******************************************************/

        // 打开信道
        void onOpenChannel(const muduo::net::TcpConnectionPtr &conn, const openChannelRequestPtr &request, muduo::Timestamp)
        {
            DEBUG_LOG("链接创建成功[%p]", conn.get());
            // 获取链接
            Connection::ptr connection = _connection_manager->getConnection(conn);
            if (connection.get() == nullptr)
            {
                DEBUG_LOG("打开信道时，不存在相应的Connection对象");
                conn->shutdown();
                return;
            }
            connection->openChannel(request);
            DEBUG_LOG("该模块正常运行");
            // 没问题
        }
        // 关闭信道
        void onCloseChannel(const muduo::net::TcpConnectionPtr &conn, const closeChannelRequestPtr &request, muduo::Timestamp)
        {
            // 获取链接
            Connection::ptr connection = _connection_manager->getConnection(conn);
            if (connection.get() == nullptr)
            {
                DEBUG_LOG("打开信道时，不存在相应的Connection对象");
                conn->shutdown();
                return;
            }
            connection->closeChannel(request);
        }
        // 声明交换机
        void onDeclareExchange(const muduo::net::TcpConnectionPtr &conn, const declareExchangeRequestPtr &request, muduo::Timestamp)
        {
            // 获取链接
            Connection::ptr connection = _connection_manager->getConnection(conn);
            if (connection.get() == nullptr)
            {
                DEBUG_LOG("声明交换机时，不存在相应的Connection对象");
                conn->shutdown();
                return;
            }
            Channel::ptr channel = connection->getChannel(request->channel_id());
            if (channel.get() == nullptr)
            {
                DEBUG_LOG("声明交换机时，不存在相应的Channel对象");
                return;
            }
            channel->declareExchange(request);
        }
        // 删除交换机
        void onDeleteExchange(const muduo::net::TcpConnectionPtr &conn, const deleteExchangeRequestPtr &request, muduo::Timestamp)
        {
            // 获取链接
            Connection::ptr connection = _connection_manager->getConnection(conn);
            if (connection.get() == nullptr)
            {
                DEBUG_LOG("删除交换机时，不存在相应的Connection对象");
                conn->shutdown();
                return;
            }
            Channel::ptr channel = connection->getChannel(request->channel_id());
            if (channel.get() == nullptr)
            {
                DEBUG_LOG("删除交换机时，不存在相应的Channel对象");
                return;
            }
            channel->deleteExchange(request);
        }
        // 声明队列
        void onDeclareQueue(const muduo::net::TcpConnectionPtr &conn, const declareQueueRequestPtr &request, muduo::Timestamp)
        {
            // 获取链接
            Connection::ptr connection = _connection_manager->getConnection(conn);
            if (connection.get() == nullptr)
            {
                DEBUG_LOG("声明队列时，不存在相应的Connection对象");
                conn->shutdown();
                return;
            }
            Channel::ptr channel = connection->getChannel(request->channel_id());
            if (channel.get() == nullptr)
            {
                DEBUG_LOG("声明队列时，不存在相应的Channel对象");
                return;
            }
            channel->declareQueue(request);
        }
        // 删除队列
        void onDeleteQueue(const muduo::net::TcpConnectionPtr &conn, const deleteQueueRequestPtr &request, muduo::Timestamp)
        {
            // 获取链接
            Connection::ptr connection = _connection_manager->getConnection(conn);
            if (connection.get() == nullptr)
            {
                DEBUG_LOG("删除队列时，不存在相应的Connection对象");
                conn->shutdown();
                return;
            }
            Channel::ptr channel = connection->getChannel(request->channel_id());
            if (channel.get() == nullptr)
            {
                DEBUG_LOG("删除队列时，不存在相应的Channel对象");
                return;
            }
            channel->deleteQueue(request);
        }
        // 队列绑定
        void onQueueBind(const muduo::net::TcpConnectionPtr &conn, const queueBindRequestPtr &request, muduo::Timestamp)
        {
            // 获取链接
            Connection::ptr connection = _connection_manager->getConnection(conn);
            if (connection.get() == nullptr)
            {
                DEBUG_LOG("绑定队列时，不存在相应的Connection对象");
                conn->shutdown();
                return;
            }
            Channel::ptr channel = connection->getChannel(request->channel_id());
            if (channel.get() == nullptr)
            {
                DEBUG_LOG("绑定队列时，不存在相应的Channel对象");
                return;
            }
            channel->bind(request);
        }
        // 队列解绑
        void onQueueUnBind(const muduo::net::TcpConnectionPtr &conn, const queueUnBindRequestPtr &request, muduo::Timestamp)
        {
            // 获取链接
            Connection::ptr connection = _connection_manager->getConnection(conn);
            if (connection.get() == nullptr)
            {
                DEBUG_LOG("解除绑定队列时，不存在相应的Connection对象");
                conn->shutdown();
                return;
            }
            Channel::ptr channel = connection->getChannel(request->channel_id());
            if (channel.get() == nullptr)
            {
                DEBUG_LOG("解除绑定队列时，不存在相应的Channel对象");
                return;
            }
            channel->unBind(request);
        }
        // 消息发布
        void onBasicPublish(const muduo::net::TcpConnectionPtr &conn, const basicPublishRequestPtr &request, muduo::Timestamp)
        {
            // 获取链接
            Connection::ptr connection = _connection_manager->getConnection(conn);
            if (connection.get() == nullptr)
            {
                DEBUG_LOG("消息发布时，不存在相应的Connection对象");
                conn->shutdown();
                return;
            }
            Channel::ptr channel = connection->getChannel(request->channel_id());
            if (channel.get() == nullptr)
            {
                DEBUG_LOG("消息发布时，不存在相应的Channel对象");
                return;
            }
            channel->basicPublish(request);
        }
        // 消息确认
        void onBasicAck(const muduo::net::TcpConnectionPtr &conn, const basicAckRequestPtr &request, muduo::Timestamp)
        {
            // 获取链接
            Connection::ptr connection = _connection_manager->getConnection(conn);
            if (connection.get() == nullptr)
            {
                DEBUG_LOG("消息确认时，不存在相应的Connection对象");
                conn->shutdown();
                return;
            }
            Channel::ptr channel = connection->getChannel(request->channel_id());
            if (channel.get() == nullptr)
            {
                DEBUG_LOG("消息确认时，不存在相应的Channel对象");
                return;
            }
            channel->basicAck(request);
        }
        // 队列消息订阅
        void onBasicConsume(const muduo::net::TcpConnectionPtr &conn, const basicConsumeRequestPtr &request, muduo::Timestamp)
        {
            // 获取链接
            Connection::ptr connection = _connection_manager->getConnection(conn);
            if (connection.get() == nullptr)
            {
                DEBUG_LOG("消息订阅时，不存在相应的Connection对象");
                conn->shutdown();
                return;
            }
            Channel::ptr channel = connection->getChannel(request->channel_id());
            if (channel.get() == nullptr)
            {
                DEBUG_LOG("消息订阅时，不存在相应的Channel对象");
                return;
            }
            channel->basicConsume(request);
        }
        // 队列消息取消订阅
        void onBasicCancel(const muduo::net::TcpConnectionPtr &conn, const basicCancelRequestPtr &request, muduo::Timestamp)
        {
            // 获取链接
            Connection::ptr connection = _connection_manager->getConnection(conn);
            if (connection.get() == nullptr)
            {
                DEBUG_LOG("消息取消订阅时，不存在相应的Connection对象");
                conn->shutdown();
                return;
            }
            Channel::ptr channel = connection->getChannel(request->channel_id());
            if (channel.get() == nullptr)
            {
                DEBUG_LOG("消息取消订阅时，不存在相应的Channel对象");
                return;
            }
            channel->basicCancel(request);
        }
        // 创建、删除、选择虚拟机
        void onDeclareHost(const muduo::net::TcpConnectionPtr &conn, const declareHostRequestPtr &request, muduo::Timestamp)
        {
            // 获取链接
            Connection::ptr connection = _connection_manager->getConnection(conn);
            if (connection.get() == nullptr)
            {
                DEBUG_LOG("消息取消订阅时，不存在相应的Connection对象");
                conn->shutdown();
                return;
            }
            Channel::ptr channel = connection->getChannel(request->channel_id());
            if (channel.get() == nullptr)
            {
                DEBUG_LOG("消息取消订阅时，不存在相应的Channel对象");
                return;
            }
            channel->declareVirtualHost(request);
        }
        void onDeleteHost(const muduo::net::TcpConnectionPtr &conn, const deleteHostRequestPtr &request, muduo::Timestamp)
        {
            // 获取链接
            Connection::ptr connection = _connection_manager->getConnection(conn);
            if (connection.get() == nullptr)
            {
                DEBUG_LOG("消息取消订阅时，不存在相应的Connection对象");
                conn->shutdown();
                return;
            }
            Channel::ptr channel = connection->getChannel(request->channel_id());
            if (channel.get() == nullptr)
            {
                DEBUG_LOG("消息取消订阅时，不存在相应的Channel对象");
                return;
            }
            channel->deleteVirtualHost(request);
        }
        void onSelectHost(const muduo::net::TcpConnectionPtr &conn, const selectHostRequestPtr &request, muduo::Timestamp)
        {
            // 获取链接
            Connection::ptr connection = _connection_manager->getConnection(conn);
            if (connection.get() == nullptr)
            {
                DEBUG_LOG("消息取消订阅时，不存在相应的Connection对象");
                conn->shutdown();
                return;
            }
            Channel::ptr channel = connection->getChannel(request->channel_id());
            if (channel.get() == nullptr)
            {
                DEBUG_LOG("消息取消订阅时，不存在相应的Channel对象");
                return;
            }
            channel->selectHost(request);
        }
        // 创建、删除用户
        void onDeclareUser(const muduo::net::TcpConnectionPtr &conn, const declareUserRequestPtr &request, muduo::Timestamp)
        {
            // 获取链接
            Connection::ptr connection = _connection_manager->getConnection(conn);
            if (connection.get() == nullptr)
            {
                DEBUG_LOG("消息取消订阅时，不存在相应的Connection对象");
                conn->shutdown();
                return;
            }
            Channel::ptr channel = connection->getChannel(request->channel_id());
            if (channel.get() == nullptr)
            {
                DEBUG_LOG("消息取消订阅时，不存在相应的Channel对象");
                return;
            }
            channel->declareUser(request);
        }
        void onDeleteUser(const muduo::net::TcpConnectionPtr &conn, const deleteUserRequestPtr &request, muduo::Timestamp)
        {
            // 获取链接
            Connection::ptr connection = _connection_manager->getConnection(conn);
            if (connection.get() == nullptr)
            {
                DEBUG_LOG("消息取消订阅时，不存在相应的Connection对象");
                conn->shutdown();
                return;
            }
            Channel::ptr channel = connection->getChannel(request->channel_id());
            if (channel.get() == nullptr)
            {
                DEBUG_LOG("消息取消订阅时，不存在相应的Channel对象");
                return;
            }
            channel->deleteUser(request);
        }
        void onSelectUser(const muduo::net::TcpConnectionPtr &conn, const selectUserRequestPtr &request, muduo::Timestamp)
        {
            // 获取链接
            Connection::ptr connection = _connection_manager->getConnection(conn);
            if (connection.get() == nullptr)
            {
                DEBUG_LOG("消息取消订阅时，不存在相应的Connection对象");
                conn->shutdown();
                return;
            }
            Channel::ptr channel = connection->getChannel(request->channel_id());
            if (channel.get() == nullptr)
            {
                DEBUG_LOG("消息取消订阅时，不存在相应的Channel对象");
                return;
            }
            channel->selectUser(request);
        }

    private:
        // 添加单例
        static BrokerServer *_instance;

        muduo::net::TcpServer _server;
        muduo::net::EventLoop _base_loop;

        ProtobufDispatcher _dispatcher;
        ProtobufCodecPtr _codec;

        ConsumerManager::ptr _consumer_manager; 
        VirtualHostManager::ptr _host_manager;
        UserManager::ptr _user_manager;
        ConnectionManager::ptr _connection_manager;
        threadpool::ptr _threadpool;
    };
}

#endif